package com.qyer.search.storm.main;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.metric.LoggingMetricsConsumer;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import com.qyer.qmara.Mara;
import com.qyer.qmara.MaraNode;
import com.qyer.qmara.NodeMode;
import com.qyer.qmara.ServiceType;
import com.qyer.search.storm.Model.ProductEntity;
import com.qyer.search.storm.Model.ProductEntity2;
import com.qyer.search.storm.bolt.FetchInfoBolt;
import com.qyer.search.storm.bolt.FetchTagBolt2;
import com.qyer.search.storm.bolt.FetchZInfoBolt;
import com.qyer.search.storm.bolt.FetchZInfoBolt2;
import com.qyer.search.storm.bolt.FetchZTagBolt;
import com.qyer.search.storm.bolt.KafkaFormatBolt;
import com.qyer.search.storm.bolt.ZIndexBolt;
import com.qyer.search.storm.bolt.ZIndexBolt2;
import com.qyer.search.storm.util.ConfigUtils;
import com.qyer.search.storm.util.Continent;
import com.qyer.search.storm.util.EntityTags;
import com.qyer.search.storm.util.Tag;
import com.qyer.search.storm.util.ZTag;
import com.qyer.search.storm.util.ZTag2;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * Created by tangzheng on 16/7/10.
 */
public class LMSV2Topology {

  public static void main(String[] args)  throws Exception{

    String[] cmdConf = args;
    Map<String, String> cmdConfs = new HashMap<>();
    /*
    ** 1:topName= @topName
    ** 2:group  = @group
    ** 3:topic  = @topic
    ** 4:version = @version
     */

    for(int j = 0; j < cmdConf.length; j++){
      String prop = cmdConf[j].trim();
      System.out.println(prop);
      String[] props = prop.split("=");
      if(props.length != 2)
        continue;
      if("topology".equals(props[0]))
        cmdConfs.put("topology", props[1]);
      if("group".equals(props[0]))
        cmdConfs.put("kafka.group.id", props[1]);
      if("topic".equals(props[0]))
        cmdConfs.put("spout.topic", props[1]);
      if("version".equals(props[0]))
        cmdConfs.put("index.version", props[1]);
    }

    if(!cmdConfs.containsKey("topology")){
      System.out.println("NO topology name!!");
      return;
    }

    Mara mara = null;
    Properties spoutProp;
    Properties tagBoltProp;
    Properties infoBoltProp;
    Properties indexBoltProp;
    Properties zTagBoltProp;

    Map spoutMap;
    Map tagBoltMap;
    Map infoBoltMap;
    Map indexBoltMap;
    Map infoBoltMapSQL;
    Map tagBoltMapSQL;
    Map zTagBoltMap;
    Map zTagBoltMapSQL;

    String serviceName = "Shizika";
    String namespace = "storm3";
    String module1 = "storm.kafkaSpout.properties";
    String module2 = "storm.fetchTagBolt.properties";
    String module3 = "storm.fetchInfoBolt.properties";
    String module4 = "storm.indexBolt.properties";
    String module5 = "storm.SQLTagBolt.properties";
    String module6 = "storm.SQLInfoBolt.properties";
    String module7 = "storm.fetchZTagBolt.properties";
    String module8 = "storm.SQLZTagBolt.properties";
    String connecting = "zk1:2181,zk2:2181,zk3:2181";
    int session = 3000;
    String root = "/disconf";


    MaraNode spout = new MaraNode(serviceName, namespace, module1, NodeMode.PERSISTENT);
    MaraNode tagBolt = new MaraNode(serviceName, namespace, module2, NodeMode.PERSISTENT);
    MaraNode infoBolt = new MaraNode(serviceName, namespace, module3, NodeMode.PERSISTENT);
    MaraNode indexBolt = new MaraNode(serviceName, namespace, module4, NodeMode.PERSISTENT);
    MaraNode tagBoltSQL = new MaraNode(serviceName, namespace, module5, NodeMode.PERSISTENT);
    MaraNode infoBoltSQL = new MaraNode(serviceName, namespace, module6, NodeMode.PERSISTENT);

    MaraNode zTagBolt = new MaraNode(serviceName, namespace, module7, NodeMode.PERSISTENT);
    MaraNode zTagBoltSQL = new MaraNode(serviceName, namespace, module8, NodeMode.PERSISTENT);


    try {
      mara = new Mara(connecting, session, root, ServiceType.DISCONF);
      spoutProp = mara.getConfProp(spout, null, -1);
      tagBoltProp = mara.getConfProp(tagBolt, null, -1);
      infoBoltProp = mara.getConfProp(infoBolt, null, -1);
      indexBoltProp = mara.getConfProp(indexBolt, null, -1);
      zTagBoltProp = mara.getConfProp(zTagBolt, null, -1);

      spoutMap = mara.getConfMap(spout, null, -1);
      tagBoltMap = mara.getConfMap(tagBolt, null, -1);
      infoBoltMap = mara.getConfMap(infoBolt, null, -1);
      indexBoltMap = mara.getConfMap(indexBolt, null, -1);
      tagBoltMapSQL = mara.getConfMap(tagBoltSQL, null, -1);
      infoBoltMapSQL = mara.getConfMap(infoBoltSQL, null, -1);
      zTagBoltMap = mara.getConfMap(zTagBolt, null, -1);
      zTagBoltMapSQL = mara.getConfMap(zTagBoltSQL, null, -1);

    }catch (Exception e){
      e.printStackTrace();
      throw e;
    }finally {
      if(mara != null)
        mara.close();
    }

    String topic = spoutProp.getProperty("spout.topic", "es_index").trim();
    String zkConnString = spoutProp.getProperty("spout.zk", "zk1:2181,zk2:2181,zk3:2181").trim();
    System.out.println("spout zk:" + zkConnString);
    //int bufsize = Integer.valueOf(spoutProp.getProperty("spout.bufsize", "2").trim());
    int parallelismSpout = Integer.valueOf(spoutProp.getProperty("spout.parallelism", "1").trim());

    String tagPrefix = tagBoltProp.getProperty("tagBolt.prefix", "tagBolt.").trim();
    int tagIncount = Integer.valueOf(tagBoltProp.getProperty("tagBolt.inCountInit", "100").trim());
    int tagIncount2 = Integer.valueOf(tagBoltProp.getProperty("tagBolt.inCountCommon", "5").trim());
    String tagStreamId = tagBoltProp.getProperty("tagBolt.streamId", "tagStream").trim();
    int parallelismTag = Integer.valueOf(tagBoltProp.getProperty("tagBolt.parallelism", "2").trim());

    String infoPrefix = infoBoltProp.getProperty("infoBolt.prefix", "infoBolt.").trim();
    int infoIncount = Integer.valueOf(
      infoBoltProp.getProperty("infoBolt.inCountInit", "100").trim());
    int infoIncount2 = Integer.valueOf(
      infoBoltProp.getProperty("infoBolt.inCountCommon", "5").trim());
    System.out.println("~~~~~~infoBolt.inCountCommon:" + infoIncount2);
    int parallelismInfo = Integer.valueOf(
      infoBoltProp.getProperty("infoBolt.parallelism", "2").trim());
    String infoStreamId = infoBoltProp.getProperty("infoBolt.streamId", "infoStream").trim();

    String indexPrefix = indexBoltProp.getProperty("indexBolt.prefix", "indexBolt.").trim();
    String version = indexBoltProp.getProperty("indexBolt.version", "default").trim();
    String serializeClass = indexBoltProp.getProperty("indexBolt.serialize", "default").trim();
    int parallelismIndex = Integer.valueOf(
      indexBoltProp.getProperty("indexBolt.parallelism", "1").trim());


    //z_tag
    String zTagPrefix = zTagBoltProp.getProperty("zTagBolt.prefix", "zTagBolt.");
    String zTagOtherPrefix = "sql.1";
    String zTagAnotherPrefix = "sql.2";
    int zTagIncount = Integer.valueOf(
      zTagBoltProp.getProperty("zTagBolt.inCountInit", "100").trim());
    int zTagIncount2 = Integer.valueOf(
      zTagBoltProp.getProperty("zTagBolt.inCountCommon", "5").trim());
    System.out.println("~~~~~~zTagBolt.inCountCommon:" + zTagIncount2);
    int parallelismZTag = Integer.valueOf(
      zTagBoltProp.getProperty("zTagBolt.parallelism", "2").trim());
    String zTagStreamId = zTagBoltProp.getProperty("zTagBolt.streamId", "zTagStream").trim();


    System.out.println("zk topic=" + topic);
    if(cmdConfs.containsKey("spout.topic")){
      topic = cmdConfs.get("spout.topic");
      System.out.println("cmd topic=" + topic);
    }
    if(cmdConfs.containsKey("index.version")){
      version = cmdConfs.get("index.version");
      System.out.println("cmd version=" + version);
    }

    String groupId = cmdConfs.get("kafka.group.id");

    System.out.println("cmd group id=" + groupId);

    BrokerHosts hosts = new ZkHosts(zkConnString);
    System.out.println("zk:"+zkConnString);
    SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, "/search/" + topic, groupId);
    spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
    spoutConfig.fetchSizeBytes = 512;
    spoutConfig.metricsTimeBucketSizeInSecs = 300;
    //spoutConfig.zkServers = Arrays.asList(new String[] {"h1", "h2", "h3"});
    //spoutConfig.zkPort = 2181;
    KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);

    TopologyBuilder builder = new TopologyBuilder();
    //parallelismSpout
    builder.setSpout("kafkaspout", kafkaSpout, 3);

    builder.setBolt("kafkaFormatBolt", new KafkaFormatBolt(), 6).shuffleGrouping("kafkaspout");

    builder.setBolt("tagBolt", new FetchTagBolt2(tagPrefix, tagIncount, tagIncount2, tagStreamId, tagBoltMapSQL), 10)
           .localOrShuffleGrouping("kafkaFormatBolt", ConfigUtils.TAGS_STREAM);
    builder.setBolt("infoBolt",
                    new FetchZInfoBolt2(infoPrefix, infoIncount, infoIncount2, infoStreamId,
                                       infoBoltMapSQL), 6)
           .localOrShuffleGrouping("kafkaFormatBolt", ConfigUtils.ENTITY_STREAM);
    builder.setBolt("zTagBolt",
                    new FetchZTagBolt(zTagPrefix, zTagIncount, zTagIncount2, zTagStreamId,
                                      zTagBoltMapSQL, zTagOtherPrefix, zTagAnotherPrefix), 6)
           .localOrShuffleGrouping("kafkaFormatBolt", ConfigUtils.ZTAG_STREAM);

    builder.setBolt("indexBolt", new ZIndexBolt2(indexPrefix, tagStreamId, infoStreamId, zTagStreamId, version, false), 16)
           .fieldsGrouping("tagBolt", tagStreamId, new Fields("msgId"))
           .fieldsGrouping("infoBolt", infoStreamId, new Fields("msgId"))
           .fieldsGrouping("zTagBolt", zTagStreamId, new Fields("msgId"));

    Config conf = new Config();
    conf.setDebug(false);
    conf.setNumAckers(1);
    conf.setNumWorkers(2);
    conf.setMessageTimeoutSecs(660);
    conf.registerMetricsConsumer(LoggingMetricsConsumer.class, 1);
    conf.registerSerialization(ProductEntity2.class);
    conf.registerSerialization(EntityTags.class);
    conf.registerSerialization(ZTag2.class);
    conf.registerSerialization(Tag.class);
    conf.registerSerialization(Continent.class);

    conf.putAll(spoutMap);
    conf.putAll(tagBoltMap);
    conf.putAll(infoBoltMap);
    conf.putAll(indexBoltMap);
    conf.putAll(zTagBoltMap);

    conf.putAll(cmdConfs);
    System.out.println("FetchZInfoBolt2");

    if (args != null && args.length > 0) {
      conf.setNumWorkers(2);

      StormSubmitter
        .submitTopologyWithProgressBar(cmdConfs.get("topology"), conf, builder.createTopology());
    }
    else {
      conf.setMaxTaskParallelism(3);

      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("search_index", conf, builder.createTopology());

      Thread.sleep(10000);

      cluster.shutdown();
    }
  }

}
